package spring.cloud.tasks.executor.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.task.TaskItemExecuteTimeoutTaskSchedulerExecutor;
import spring.cloud.tasks.executor.utils.ExecutorServiceUtil;
import spring.cloud.tasks.executor.utils.SystemEnvProperties;
import spring.cloud.tasks.executor.utils.ThreadFactory;
import spring.cloud.tasks.executor.utils.log.LogEvents;
import spring.cloud.tasks.executor.utils.log.LogUtils;
import spring.cloud.tasks.executor.utils.zookeeper.NameSpaceConfiguration;
import spring.cloud.tasks.executor.utils.zookeeper.Zookeeper;
import spring.cloud.tasks.executor.utils.zookeeper.ZookeeperImpl;

import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Setter
@Getter
public class Executor {
    private static AtomicBoolean inited = new AtomicBoolean(false);
    private String nameSpace;
    private String executorId;
    private Zookeeper zookeeper;
    private ConnectionStateListener connectionStateListener;
    private ExecutorAndTaskInitService executorAndTaskInitService;
    //
    private ReentrantLock lock = new ReentrantLock();
    private Thread restartThread;
    private volatile boolean needRestart = false;
    private volatile boolean isShutdown;
    //
    private ExecutorResetCountService executorResetCountService;

    private Executor(String nameSpace, String executorId) {
        this.nameSpace = nameSpace;
        this.executorId = executorId;
        initRestartThread();
    }

    public static Executor buildExecutor(String namespace, String executorName) {
        if ("$self".equals(namespace)) {
            throw new RuntimeException("The namespace cannot be $SaturnSelf");
        }
        if (!inited.compareAndSet(false, true)) {
            return null;
        }
        return new Executor(namespace, executorName);
    }

    private void initRestartThread() {
        final String threadName = executorId + "-restart-thread";
        this.restartThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while (true) {
                        if (isShutdown) {
                            return;
                        }
                        if (needRestart) {
                            try {
                                needRestart = false;
                                //
                                execute();
                            } catch (InterruptedException e) {
                                throw e;
                            } catch (Throwable t) {
                                needRestart = true;
                                LogUtils.error(log, LogEvents.ExecutorEvent.REINIT, "Executor {} reinitialize failed, will retry again", executorId, t);
                            }
                        }
                        Thread.sleep(1000L);
                    }
                } catch (InterruptedException e) {
                    LogUtils.info(log, LogEvents.ExecutorEvent.REINIT, "{} is interrupted", threadName);
                    Thread.currentThread().interrupt();
                }
            }
        });
        this.restartThread.setName(threadName);
        this.restartThread.setDaemon(false);
        this.restartThread.start();
    }


    public void execute() throws Exception {
        lock.lockInterruptibly();
        try {
            if (isShutdown) {
                return;
            }
            long startTime = System.currentTimeMillis();
            shutdown0();
            try {
                LogUtils.info(log, LogEvents.ExecutorEvent.INIT, "start to discover from saturn console");
                String connectString = null;//配置中心获取
                if (connectString == null) {
                    LogUtils.error(log, LogEvents.ExecutorEvent.INIT, "zk connection string is blank!");
                    throw new RuntimeException("zk connection string is blank!");
                }
                // 初始化注册中心
                initZookeeper(connectString);
                // 注册Executor
                registerExecutor();
                // 启动零点清0成功数错误数的线程
                LogUtils.info(log, LogEvents.ExecutorEvent.INIT, "start ResetCountService");
                executorResetCountService = new ExecutorResetCountService(executorId);
                executorResetCountService.start();
                //
                // 初始化timeout scheduler
                LogUtils.info(log, LogEvents.ExecutorEvent.INIT, "start to create timeout scheduler");
                TaskItemExecuteTimeoutTaskSchedulerExecutor.createScheduledThreadPoolExecutor(executorId);
                // 添加新增作业时的回调方法，启动已经存在的作业
                LogUtils.info(log, LogEvents.ExecutorEvent.INIT, "start to register newJobCallback, and async start existing jobs");
                executorAndTaskInitService.registerTaskWatcher();
                //
                LogUtils.info(log, LogEvents.ExecutorEvent.INIT, "The executor {} start successfully which used {} ms", executorId, System.currentTimeMillis() - startTime);
            } catch (Throwable t) {
                shutdown0();
                throw t;
            }
        } finally {
            lock.unlock();
        }
    }

    private void initZookeeper(String connectString) throws Exception {
        try {
            // 初始化注册中心
            LogUtils.info(log, LogEvents.ExecutorEvent.INIT, "start to init zookeeper");
            NameSpaceConfiguration nameSpaceConfiguration = new NameSpaceConfiguration(connectString, nameSpace, 1000, 3000);
            zookeeper = new ZookeeperImpl(nameSpaceConfiguration);
            zookeeper.init();
            connectionStateListener = new ConnectionStateListener(executorId) {
                @Override
                public void onLost() {
                    needRestart = true;
                }
            };
            zookeeper.addConnectionStateListener(connectionStateListener);
            executorAndTaskInitService = new ExecutorAndTaskInitService(executorId, zookeeper);
        } catch (Exception e) {
            throw e;
        }
    }

    private void registerExecutor() throws Exception {
        try {
            LogUtils.info(log, LogEvents.ExecutorEvent.INIT, "start to register executor");
            executorAndTaskInitService.registerExecutor();
        } catch (Exception e) {
            LogUtils.error(log, LogEvents.ExecutorEvent.INIT, "register executor error", e);
            throw e;
        }
    }


    public void shutdown() throws Exception {
        if (isShutdown) {
            return;
        }
        lock.lockInterruptibly();
        try {
            if (isShutdown) {
                return;
            }
            shutdown0();
            restartThread.interrupt();
            isShutdown = true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * Executor优雅退出： 把自己从集群中拿掉，现有的作业不停； 一直到全部作业都执行完毕，再真正退出； 设置一定超时时间，如果超过这个时间仍未退出，则强行中止
     */
    public void shutdownGracefully() throws Exception {
        if (isShutdown) {
            return;
        }
        lock.lockInterruptibly();
        try {
            if (isShutdown) {
                return;
            }
            shutdownGracefully0();
            restartThread.interrupt();
            //
            isShutdown = true;
        } finally {
            lock.unlock();
        }
    }

    private void shutdownGracefully0() throws Exception {
        lock.lockInterruptibly();
        try {
            LogUtils.info(log, LogEvents.ExecutorEvent.GRACEFUL_SHUTDOWN, "Try to stop executor {} gracefully", executorId);
            if (executorAndTaskInitService != null) {
                executorAndTaskInitService.unregisterJobsWatcher();
            }
            // 先关闭统计信息上报，因为上报的zk结点为servers/xxx/xxx，如果放在下线后再关闭，则导致ExecutorCleanService执行后，仍然在上报统计信息，垃圾结点由此而生
            shutdownTaskExecutorCountThread();
            // shutdown timeout-watchdog-threadpool
            TaskItemExecuteTimeoutTaskSchedulerExecutor.shutdownScheduledThreadPoolExecutor(executorId);
            blockUntilTaskCompletedIfNotTimeout();
            shutdownTaskExecutors();
            ExecutorIdToTaskIdToTaskExecutorMapMap.removeTaskExecutor(executorId);
            if (executorResetCountService != null) {
                executorResetCountService.stop();
            }
            if (executorAndTaskInitService != null) {
                executorAndTaskInitService.unRegisterExecutor();
            }
            if (connectionStateListener != null) {
                connectionStateListener.close();
            }
            if (zookeeper != null) {
                zookeeper.close();
            }
            LogUtils.info(log, LogEvents.ExecutorEvent.GRACEFUL_SHUTDOWN, "executor {} is stopped gracefully", executorId);
        } finally {
            lock.unlock();
        }
    }


    /**
     * Executor关闭
     */
    private void shutdown0() throws Exception {
        lock.lockInterruptibly();
        try {
            LogUtils.info(log, LogEvents.ExecutorEvent.SHUTDOWN, "Try to stop executor {}", executorId);
            if (executorAndTaskInitService != null) {
                executorAndTaskInitService.unregisterJobsWatcher();
            }
            // 先关闭统计信息上报，因为上报的zk结点为servers/xxx/xxx，如果放在下线后再关闭，则导致ExecutorCleanService执行后，仍然在上报统计信息，垃圾结点由此而生
            shutdownTaskExecutorCountThread();
            TaskItemExecuteTimeoutTaskSchedulerExecutor.shutdownScheduledThreadPoolExecutor(executorId);
            shutdownTaskExecutors();
            ExecutorIdToTaskIdToTaskExecutorMapMap.removeTaskExecutor(executorId);


            //
            if (executorResetCountService != null) {
                executorResetCountService.stop();
            }
            if (executorAndTaskInitService != null) {
                executorAndTaskInitService.unRegisterExecutor();
            }
            if (connectionStateListener != null) {
                connectionStateListener.close();
            }
            if (zookeeper != null) {
                zookeeper.close();
            }
            LogUtils.info(log, LogEvents.ExecutorEvent.SHUTDOWN, "The executor {} is stopped", executorId);
        } finally {
            lock.unlock();
        }
    }

    private void shutdownTaskExecutorCountThread() {
        Map<String, TaskExecutor> taskIdToTaskExecutorMap = ExecutorIdToTaskIdToTaskExecutorMapMap.getExecutorIdToTaskIdToTaskExecutorMapMap().get(executorId);
        if (taskIdToTaskExecutorMap != null) {
            Iterator<String> iterator = taskIdToTaskExecutorMap.keySet().iterator();
            while (iterator.hasNext()) {
                String taskId = iterator.next();
                TaskExecutor taskExecutor = taskIdToTaskExecutorMap.get(taskId);
                if (taskExecutor != null) {
                    taskExecutor.shutdownCountThread();
                }
            }
        }
    }

    /**
     * block until all Job is completed if it is not timeout
     */
    private void blockUntilTaskCompletedIfNotTimeout() {
        Map<String, TaskExecutor> taskIdToTaskExecutorMap = ExecutorIdToTaskIdToTaskExecutorMapMap.getExecutorIdToTaskIdToTaskExecutorMapMap().get(executorId);
        if (taskIdToTaskExecutorMap == null) {
            return;
        }
        Set<Entry<String, TaskExecutor>> entrySet = taskIdToTaskExecutorMap.entrySet();
        if (entrySet == null || entrySet.size() == 0) {
            return;
        }
        long startTime = System.currentTimeMillis();
        boolean hasRunning = false;
        do {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                LogUtils.error(log, LogEvents.ExecutorEvent.GRACEFUL_SHUTDOWN, e.getMessage(), e);
                Thread.currentThread().interrupt();
            }
            //
            for (Entry<String, TaskExecutor> entry : entrySet) {
                TaskExecutor taskExecutor = entry.getValue();
                if (taskExecutor.isAllowedShutdownGracefully()) {
                    if (taskExecutor.getTaskExecuteEngine().isRunning()) {
                        hasRunning = true;
                        break;
                    } else {
                        hasRunning = false;
                    }
                }
                // 其他作业（消息作业）不等，因为在接下来的forceStop是优雅强杀的，即等待一定时间让业务执行再强杀
            }
        } while (hasRunning && System.currentTimeMillis() - startTime < SystemEnvProperties.VIP_SATURN_SHUTDOWN_TIMEOUT * 1000l);
        LogUtils.info(log, LogEvents.ExecutorEvent.GRACEFUL_SHUTDOWN, "Shutdown phase [blockUntilJobCompletedIfNotTimeout] took {}ms", System.currentTimeMillis() - startTime);
    }

    private void shutdownTaskExecutors() {
        Map<String, TaskExecutor> taskIdToTaskExecutorMap = ExecutorIdToTaskIdToTaskExecutorMapMap.getExecutorIdToTaskIdToTaskExecutorMapMap().get(executorId);
        if (taskIdToTaskExecutorMap == null) {
            return;
        }
        long startTime = System.currentTimeMillis();
        Iterator<String> iterator = taskIdToTaskExecutorMap.keySet().iterator();
        ExecutorService executorService = Executors.newCachedThreadPool(new ThreadFactory(executorId + "-shutdownTaskExecutors-thread", true));
        List<Future<?>> futureList = new ArrayList<>();
        while (iterator.hasNext()) {
            final String taskId = iterator.next();
            final TaskExecutor taskExecutor = taskIdToTaskExecutorMap.get(taskId);
            if (taskExecutor != null) {
                futureList.add(executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            taskExecutor.shutdown(false);
                        } catch (Throwable t) {
                            LogUtils.error(log, taskId, "shutdown JobScheduler error", t);
                        }
                    }
                }));
            }
        }
        for (Future<?> future : futureList) {
            try {
                future.get();
            } catch (Exception e) {
                LogUtils.error(log, LogEvents.ExecutorEvent.SHUTDOWN, "wait shutdown job error", e);
            }
        }
        ExecutorServiceUtil.shutdownNow(executorService);
        LogUtils.info(log, LogEvents.ExecutorEvent.SHUTDOWN, "Shutdown phase [shutdownJobSchedulers] took {}ms", System.currentTimeMillis() - startTime);
    }


}
